#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/TcpConnection.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/threadpool.hpp"
#include "consumer.hpp"
#include "host.hpp"
#include "routing.hpp"

namespace mmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using OpenChannelRequestPtr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<CloseChannelRequest>;
    using DeclareExchangeRequestPtr = std::shared_ptr<DeclareExchangeRequest>;
    using DeleteExchangeRequestPtr = std::shared_ptr<DeleteExchangeRequest>;
    using DeclareQueueRequestPtr = std::shared_ptr<DeclareQueueRequest>;
    using DeleteQueueRequestPtr = std::shared_ptr<DeleteQueueRequest>;
    using QueueBindRequestPtr = std::shared_ptr<QueueBindRequest>;
    using QueueUnBindRequestPtr = std::shared_ptr<QueueUnBindRequest>;
    using BasicPublishRequestPtr = std::shared_ptr<BasicPublishRequest>;
    using BasicAckRequestPtr = std::shared_ptr<BasicAckRequest>;
    using BasicConsumeRequestPtr = std::shared_ptr<BasicConsumeRequest>;
    using BasicCancelRequestPtr = std::shared_ptr<BasicCancelRequest>;

    // 一个管道对应一个消费者
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    public:
        Channel(const std::string &id, 
            const ConsumerManager::ptr &cmp,
            const VirtualHost::ptr &host, 
            const ProtobufCodecPtr &codec, 
            const muduo::net::TcpConnectionPtr &conn,
            const Threadpool::ptr &pool)
            : _id(id), _conn(conn), _codec(codec), _cmp(cmp), _host(host), _pool(pool)
        {
            DLOG("new Channel: %p", this);
        }
        ~Channel()
        {
            if(_consumer.get() != nullptr)
            {
                _cmp->Remove(_consumer->_tag, _consumer->_qname);
            }
            DLOG("del Channel: %p", this);
        }

        // 交换机的声明与删除
        void DeclareExchange(const DeclareExchangeRequestPtr &req)
        {
            bool ret = _host->DeclareExchange(req->exchange_name(), req->exchange_type(), req->durable(),
                req->auto_delete(), req->args());
            BasicResponse(req->rid(), req->cid(), ret);
        }
        void DeleteExchange(const DeleteExchangeRequestPtr &req)
        {
            _host->DeleteExchange(req->exchange_name());
            BasicResponse(req->rid(), req->cid(), true);
        }
        // 队列的声明与删除
        void DeclareQueue(const DeclareQueueRequestPtr &req)
        {
            bool ret = _host->DeclareQueue(req->queue_name(), req->durable(), req->exclusive(),
                req->auto_delete(), req->args());
            if(ret == false) 
                return BasicResponse(req->rid(), req->cid(), false);
            _cmp->InitQueueConsumer(req->queue_name()); // 初始化队列的消费者管理句柄
            BasicResponse(req->rid(), req->cid(), true);
        }
        void DeleteQueue(const DeleteQueueRequestPtr &req)
        {
            _cmp->DestroyQueueConsumer(req->queue_name());
            _host->DeleteQueue(req->queue_name());
            BasicResponse(req->rid(), req->cid(), true);
        }
        // 队列的绑定与解绑
        void QueueBind(const QueueBindRequestPtr &req)
        {
            bool ret = _host->Bind(req->exchange_name(), req->queue_name(), req->binding_key());
            BasicResponse(req->rid(), req->cid(), ret);
        }
        void QueueUnBind(const QueueUnBindRequestPtr &req)
        {
            _host->UnBind(req->exchange_name(), req->queue_name());
            BasicResponse(req->rid(), req->cid(), true);
        }
        // 消息的发布
        void BasicPublish(const BasicPublishRequestPtr &req)
        {
            // 1. 判断交换机是否存在
            Exchange::ptr ep = _host->SelectExchange(req->exchange_name());
            if(ep.get() == nullptr)
            {
                return BasicResponse(req->rid(), req->cid(), false);
            }
            // 2. 进行路由交换，判断消息可以发送到交换机绑定的哪个队列中
            MsgQueueBindingMap mqbm = _host->ExchangeBindings(req->exchange_name());
            BasicProperties *properties = nullptr;
            std::string routing_key;
            if(req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for(auto &binding : mqbm)
            {
                // 匹配不成功的消息将被丢弃（这里简单设计，进一步可以设计为交给备份交换机处理这些消息）
                if(Routing::Route(ep->_type, routing_key, binding.second->_binding_key))
                {
                    // 3. 将消息添加到队列中(添加消息的管理)
                    _host->BasicPublish(binding.first, properties, req->body());
                    // 4. 若该队列存在消费者则向线程池中添加一个消息消费任务(由线程池完成向指定队列的订阅者推送消息)
                    if(_cmp->ConsumerCount(binding.first) > 0)
                    {
                        auto task = std::bind(&Channel::Consume, this, binding.first);
                        _pool->Push(task);
                    }
                }
            }
            BasicResponse(req->rid(), req->cid(), true);
        }
        // 消息的确认
        void BasicAck(const BasicAckRequestPtr &req)
        {
            _host->BasicAck(req->queue_name(), req->message_id());
            BasicResponse(req->rid(), req->cid(), true);
        }
        // 订阅队列消息
        void BasicConsume(const BasicConsumeRequestPtr &req)
        {
            // 1. 判断队列是否存在
            bool ret = _host->ExistQueue(req->queue_name());
            if(ret == false)
            {
                return BasicResponse(req->rid(), req->cid(), false);
            }
            // 2. 创建队列消费者
            auto task = std::bind(&Channel::Callback, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3);
            // 创建消费者后，当前的channel角色就是消费者
            _consumer = _cmp->Create(req->consumer_tag(), req->queue_name(), req->auto_ack(), task);
            BasicResponse(req->rid(), req->cid(), true);
            // 3. 若存在历史消息或在创建消费者之前发布消息，需要在队列中第一个消费者创建时把对应数量的消息任务抛入线程池
            if(_cmp->ConsumerCount(req->queue_name()) == 1)
            {
                auto task = std::bind(&Channel::Consume, this, req->queue_name());
                size_t num = _host->GetQueueAbleMessageCount(req->queue_name());
                // DLOG("向线程池抛入%ld个任务", _host->GetQueueAbleMessageCount(req->queue_name()));
                while(num--)
                {
                    _pool->Push(task);
                }
            }
        }
        // 取消订阅
        void BasicCancel(const BasicCancelRequestPtr &req)
        {
            _cmp->Remove(req->consumer_tag(), req->queue_name());
            BasicResponse(req->rid(), req->cid(), true);
        }
    private:
        // 组织推送消息，将消息推送给channel对应的客户端
        void Callback(const std::string tag, const BasicProperties *bp, const std::string body)
        {
            BasicConsumeResponse resp;
            resp.set_cid(_id);
            resp.set_consumer_tag(_consumer->_tag);
            resp.set_body(body);
            if(bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
        // 指定队列消费消息
        void Consume(const std::string &qname)
        {
            // 1. 从队列中获取一个消息
            MessagePtr mp = _host->BasicConsume(qname);
            if(mp.get() == nullptr)
            {
                DLOG("执行消费任务失败，%s 队列没有消息", qname.c_str());
                return;
            }
            // 2. 从队列中获取一个订阅者
            Consumer::ptr cp = _cmp->Choose(qname);
            if(cp.get() == nullptr)
            {
                // DLOG("执行消费任务失败，%s 队列没有消费者", qname.c_str());
                return;
            }
            // 3. 调用订阅者对应的消息处理函数，实现消息的推送
            cp->_callback(cp->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            // 4. 判断订阅者是否是自动确认----不需要等待确认，直接删除消息，否则需要外部收到消息确认后再删除
            if(cp->_auto_ack)
                _host->BasicAck(qname, mp->payload().properties().id());
        }

        void BasicResponse(const std::string &rid, const std::string &cid, bool ok)
        {
            BasicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
    private:
        std::string _id;
        Consumer::ptr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ConsumerManager::ptr _cmp;
        VirtualHost::ptr _host;
        Threadpool::ptr _pool;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
    public:
        ChannelManager()
        {}

        bool OpenChannel(const std::string &id, 
            const ConsumerManager::ptr &cmp,
            const VirtualHost::ptr &host, 
            const ProtobufCodecPtr &codec, 
            const muduo::net::TcpConnectionPtr &conn,
            const Threadpool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if(it != _channels.end())
            {
                return false;
            }
            Channel::ptr channel = std::make_shared<Channel>(id, cmp, host, codec, conn, pool);
            _channels.insert({id, channel});
            return true;
        }
        void CloseChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }
        Channel::ptr GetChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
    
}


#endif